package github.sf.fw.redis;

import com.fasterxml.jackson.databind.ObjectMapper;
import github.sf.fw.utils.StrUtil;
import org.redisson.Redisson;
import org.redisson.api.RFuture;
import org.redisson.api.RedissonClient;
import org.redisson.client.RedisConnection;
import org.redisson.client.protocol.RedisCommands;
import org.redisson.codec.JsonJacksonCodec;
import org.redisson.codec.SerializationCodec;
import org.redisson.config.Config;
import org.redisson.config.SingleServerConfig;

import java.util.concurrent.*;
import java.util.concurrent.atomic.AtomicInteger;

/**
 * 使用Redisson连接Redis
 */
public class RedisUtil {
    private final RedissonClient redissonClient;
    private final ExecutorService executorService = Executors.newFixedThreadPool(3, new ThreadFactory() {
        private final AtomicInteger count = new AtomicInteger(0);

        @Override
        public Thread newThread(Runnable r) {
            return new Thread(r, "redisson-thread-" + count.incrementAndGet());
        }
    });

    public RedisUtil(String address, String password) {
        Config config = initPartRedisConfig(address, password);
        config.setCodec(new JsonJacksonCodec());
        this.redissonClient = Redisson.create(config);
    }

    public RedisUtil(String address, String password, ObjectMapper objectMapper) {
        Config config = initPartRedisConfig(address, password);
        config.setCodec(new JsonJacksonCodec(objectMapper));
        this.redissonClient = Redisson.create(config);
    }

    /**
     * 初始化部分redis 配置
     */
    private Config initPartRedisConfig(String address, String password) {
        Config config = new Config();
        SingleServerConfig singleServerConfig = config
            .useSingleServer()
            .setAddress(address)
            .setTimeout(5000)
            .setConnectionPoolSize(10);
        if (!StrUtil.isEmpty(password)) {
            singleServerConfig.setPassword(password);
        }
        config.setExecutor(executorService);
        return config;
    }

    public void set(RedisConnection redisConn, String key, Object value) {
        redisConn.sync(new SerializationCodec(), RedisCommands.SET, key, value);
    }

    public Object get(String key) {
        return redissonClient.getBucket(key).get();
    }

    public Object poll(String key) {
        return redissonClient.getBlockingQueue(key).poll();
    }

    public void pushSync(String key, Object value) {
        RFuture<Void> voidRFuture = redissonClient.getBlockingQueue(key).putAsync(value);
        try {
            //wait push value
            voidRFuture.get();
        } catch (InterruptedException | ExecutionException e) {
            throw new RuntimeException(e);
        }
    }

    public void set(String key, Object value) {
        redissonClient.getBucket(key).set(value);
    }

    public boolean deleteBucket(String key) {
        return redissonClient.getBucket(key).delete();
    }

    public boolean deleteQueue(String key) {
        if (redissonClient.getBlockingQueue(key).isExists()) {
            return redissonClient.getBlockingQueue(key).delete();
        }
        return true;
    }

    public RedissonClient getRedissonClient() {
        return redissonClient;
    }

    public void close() {
        executorService.shutdown();
        if (!executorService.isShutdown()) {
            executorService.shutdownNow();
        }
        if (redissonClient != null) {
            redissonClient.shutdown(1, 3, TimeUnit.SECONDS);
        }
    }
}
